

package com.coding4play.mqtt.broker.protocol;

import com.coding4play.mqtt.broker.config.BrokerProperties;
import com.coding4play.mqtt.broker.internal.AggregatedSink;
import com.coding4play.mqtt.broker.internal.InternalMessage;
import com.coding4play.mqtt.common.entity.message.DupPublishMessage;
import com.coding4play.mqtt.common.entity.message.RetainMessage;
import com.coding4play.mqtt.common.entity.message.SubMessage;
import com.coding4play.mqtt.common.entity.session.ChannelSession;
import com.coding4play.mqtt.common.repository.*;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

/**
 * PUBLISH连接处理
 */
@Slf4j
@AllArgsConstructor
public class Publish {

    private ISessionRepository sessionRepository;
    private ISubMessageRepository subMessageRepository;
    private IMessageIdService<Integer> messageIdService;
    private IRetainMessageRepository retainMessageRepository;
    private IDupPublishMessageRepository dupPublishMessageRepository;
    private AggregatedSink aggregatedSink;
    private BrokerProperties brokerProperties;


    public void processPublish(Channel channel, MqttPublishMessage msg) {
        String clientId = (String) channel.attr(AttributeKey.valueOf("clientId")).get();
        // publish 延长session失效时间
        if (sessionRepository.hasSession(clientId)) {
            ChannelSession channelSession = sessionRepository.get(clientId);
//            ChannelId channelId = channelIdMap.get(sessionStore.getBrokerId() + "_" + sessionStore.getChannelId());
            Channel c0 = this.sessionRepository.getChannel(channelSession);
            if (brokerProperties.getId().equals(channelSession.getBrokerId()) && c0 != null) {
                sessionRepository.expire(clientId, channelSession.getExpire());
            }
        }
        // QoS=0
        if (msg.fixedHeader().qosLevel() == MqttQoS.AT_MOST_ONCE) {
            byte[] messageBytes = new byte[msg.payload().readableBytes()];
            msg.payload().getBytes(msg.payload().readerIndex(), messageBytes);
            InternalMessage internalMessage = InternalMessage.builder()
                    .topic(msg.variableHeader().topicName())
                    .mqttQoS(msg.fixedHeader().qosLevel().value())
                    .messageBytes(messageBytes)
                    .dup(false)
                    .retain(false)
                    .clientId(clientId)
                    .build();
            aggregatedSink.sink(internalMessage);
            this.sendPublishMessage(msg.variableHeader().topicName(), msg.fixedHeader().qosLevel(), messageBytes, false, false);
        }
        // QoS=1
        if (msg.fixedHeader().qosLevel() == MqttQoS.AT_LEAST_ONCE) {
            byte[] messageBytes = new byte[msg.payload().readableBytes()];
            msg.payload().getBytes(msg.payload().readerIndex(), messageBytes);
            InternalMessage internalMessage = InternalMessage.builder()
                    .topic(msg.variableHeader().topicName())
                    .mqttQoS(msg.fixedHeader().qosLevel().value())
                    .messageBytes(messageBytes)
                    .dup(false)
                    .retain(false)
                    .clientId(clientId)
                    .build();
            aggregatedSink.sink(internalMessage);
            this.sendPublishMessage(msg.variableHeader().topicName(), msg.fixedHeader().qosLevel(), messageBytes, false, false);
            this.sendPubAckMessage(channel, msg.variableHeader().packetId());
        }
        // QoS=2
        if (msg.fixedHeader().qosLevel() == MqttQoS.EXACTLY_ONCE) {
            byte[] messageBytes = new byte[msg.payload().readableBytes()];
            msg.payload().getBytes(msg.payload().readerIndex(), messageBytes);
            InternalMessage internalMessage = InternalMessage.builder()
                    .topic(msg.variableHeader().topicName())
                    .mqttQoS(msg.fixedHeader().qosLevel().value())
                    .messageBytes(messageBytes)
                    .dup(false)
                    .retain(false)
                    .clientId(clientId)
                    .build();
            aggregatedSink.sink(internalMessage);
            this.sendPublishMessage(msg.variableHeader().topicName(), msg.fixedHeader().qosLevel(), messageBytes, false, false);
            this.sendPubRecMessage(channel, msg.variableHeader().packetId());
        }
        // retain=1, 保留消息
        if (msg.fixedHeader().isRetain()) {
            byte[] messageBytes = new byte[msg.payload().readableBytes()];
            msg.payload().getBytes(msg.payload().readerIndex(), messageBytes);
            if (messageBytes.length == 0) {
                retainMessageRepository.delete(msg.variableHeader().topicName());
            } else {
                RetainMessage retainMessage = RetainMessage.builder()
                        .topic(msg.variableHeader().topicName())
                        .mqttQoS(msg.fixedHeader().qosLevel().value())
                        .messageBytes(messageBytes)
                        .build();
                retainMessageRepository.put(msg.variableHeader().topicName(), retainMessage);
            }
        }
    }

    private void sendPublishMessage(String topic, MqttQoS mqttQoS, byte[] messageBytes, boolean retain, boolean dup) {
        List<SubMessage> subMessages = subMessageRepository.search(topic);
        subMessages.forEach(subscribeStore -> {
            if (sessionRepository.hasSession(subscribeStore.getClientId())) {
                // 订阅者收到MQTT消息的QoS级别, 最终取决于发布消息的QoS和主题订阅的QoS
                MqttQoS respQoS = mqttQoS.value() > subscribeStore.getMqttQoS() ? MqttQoS.valueOf(subscribeStore.getMqttQoS()) : mqttQoS;
                if (respQoS == MqttQoS.AT_MOST_ONCE) {
                    MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                            new MqttFixedHeader(MqttMessageType.PUBLISH, dup, respQoS, retain, 0),
                            new MqttPublishVariableHeader(topic, 0), Unpooled.buffer().writeBytes(messageBytes));
                    log.debug("PUBLISH - clientId: {}, topic: {}, Qos: {}", subscribeStore.getClientId(), topic, respQoS.value());
                    ChannelSession channelSession = sessionRepository.get(subscribeStore.getClientId());
                    Channel c0 = this.sessionRepository.getChannel(channelSession);
                    if (c0 != null) {
                        c0.writeAndFlush(publishMessage);
                    }
                }
                if (respQoS == MqttQoS.AT_LEAST_ONCE) {
                    int messageId = messageIdService.nextMessageId();
                    MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                            new MqttFixedHeader(MqttMessageType.PUBLISH, dup, respQoS, retain, 0),
                            new MqttPublishVariableHeader(topic, messageId), Unpooled.buffer().writeBytes(messageBytes));
                    log.debug("PUBLISH - clientId: {}, topic: {}, Qos: {}, messageId: {}", subscribeStore.getClientId(), topic, respQoS.value(), messageId);
                    DupPublishMessage dupPublishMessage = DupPublishMessage.builder()
                            .clientId(subscribeStore.getClientId())
                            .topic(topic)
                            .mqttQoS(respQoS.value())
                            .messageBytes(messageBytes)
                            .messageId(messageId)
                            .build();
                    dupPublishMessageRepository.put(subscribeStore.getClientId(), dupPublishMessage);
                    ChannelSession channelSession = sessionRepository.get(subscribeStore.getClientId());
                    Channel c0 = this.sessionRepository.getChannel(channelSession);
                    if (c0 != null) {
                        c0.writeAndFlush(publishMessage);
                    }
                }
                if (respQoS == MqttQoS.EXACTLY_ONCE) {
                    int messageId = messageIdService.nextMessageId();
                    MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                            new MqttFixedHeader(MqttMessageType.PUBLISH, dup, respQoS, retain, 0),
                            new MqttPublishVariableHeader(topic, messageId), Unpooled.buffer().writeBytes(messageBytes));
                    log.debug("PUBLISH - clientId: {}, topic: {}, Qos: {}, messageId: {}", subscribeStore.getClientId(), topic, respQoS.value(), messageId);
                    DupPublishMessage dupPublishMessage = DupPublishMessage.builder()
                            .clientId(subscribeStore.getClientId())
                            .topic(topic)
                            .mqttQoS(respQoS.value())
                            .messageBytes(messageBytes)
                            .messageId(messageId)
                            .build();
                    dupPublishMessageRepository.put(subscribeStore.getClientId(), dupPublishMessage);
                    ChannelSession channelSession = sessionRepository.get(subscribeStore.getClientId());
                    Channel c0 = this.sessionRepository.getChannel(channelSession);
                    if (c0 != null) {
                        c0.writeAndFlush(publishMessage);
                    }
                }
            }
        });
    }

    private void sendPubAckMessage(Channel channel, int messageId) {
        MqttPubAckMessage pubAckMessage = (MqttPubAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(messageId), null);
        channel.writeAndFlush(pubAckMessage);
    }

    private void sendPubRecMessage(Channel channel, int messageId) {
        MqttMessage pubRecMessage = MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(messageId), null);
        channel.writeAndFlush(pubRecMessage);
    }

}
